package com.flink.hbase.multithreaded;

import com.flink.hbase.User;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.format.DateTimeFormatter;
import java.util.Map;

/**
 * 基于多线程的 User 数据 Sink
 * 继承 HBaseMultithreadedSink，专门用于写入 User 数据到 HBase
 */
public class UserMultithreadedSink extends HBaseMultithreadedSink<User> {
    private static final Logger LOG = LoggerFactory.getLogger(UserMultithreadedSink.class);
    
    private static final String COLUMN_NAME = "name";
    private static final String COLUMN_AGE = "age";
    private static final String COLUMN_GENDER = "gender";
    private static final String COLUMN_TIMESTAMP = "timestamp";
    private static final DateTimeFormatter DATETIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    
    public UserMultithreadedSink(String tableName, String columnFamily, 
                               int batchSize, int asyncThreads, 
                               long flushInterval, Map<String, String> hbaseConfig) {
        super(tableName, columnFamily, batchSize, asyncThreads, flushInterval, hbaseConfig);
    }
    
    @Override
    protected Put convertToPut(User user) throws Exception {
        if (user == null || user.getUserId() == null || user.getUserId().isEmpty()) {
            LOG.warn("Invalid user record, skipping: {}", user);
            return null;
        }
        
        try {
            // 使用 userId 作为 rowkey
            byte[] rowkey = Bytes.toBytes(user.getUserId());
            Put put = new Put(rowkey);
            
            // 获取列族字节数组
            byte[] cfBytes = Bytes.toBytes(getColumnFamily());
            
            // 添加各个字段
            if (user.getName() != null) {
                put.addColumn(cfBytes, Bytes.toBytes(COLUMN_NAME), Bytes.toBytes(user.getName()));
            }
            
            put.addColumn(cfBytes, Bytes.toBytes(COLUMN_AGE), Bytes.toBytes(String.valueOf(user.getAge())));
            
            if (user.getGender() != null) {
                put.addColumn(cfBytes, Bytes.toBytes(COLUMN_GENDER), Bytes.toBytes(user.getGender()));
            }
            
            if (user.getTimestamp() != null) {
                String timestampStr = user.getTimestamp().format(DATETIME_FORMATTER);
                put.addColumn(cfBytes, Bytes.toBytes(COLUMN_TIMESTAMP), Bytes.toBytes(timestampStr));
            }
            
            return put;
            
        } catch (Exception e) {
            LOG.error("Error converting user to Put: {}", user, e);
            throw e;
        }
    }
    
    /**
     * 工厂方法，创建 UserMultithreadedSink 实例
     */
    public static UserMultithreadedSink create(String tableName, String columnFamily, 
                                             int batchSize, int asyncThreads, 
                                             long flushInterval, Map<String, String> hbaseConfig) {
        return new UserMultithreadedSink(tableName, columnFamily, batchSize, 
                                       asyncThreads, flushInterval, hbaseConfig);
    }
} 